package com.gupao.vip.trevor.queue;

import com.gupao.vip.trevor.queue.zookeeper.ZKSession;
import com.gupao.vip.trevor.queue.zookeeper.ZKSessionFactory;

import javax.naming.OperationNotSupportedException;
import java.io.*;
import java.util.AbstractQueue;
import java.util.Iterator;

/**
 * <h1> 分布式队列 </h1>
 * <p>
 * <b>描述：</b>
 *        队列方法调用ZooKeeper实现，需要预先在ZooKeeper队列中
 *        的根路径创建好queue节点。队列元素会被以有序节点的形式存
 *        储到queue节点下。分布式队列支持并发访问，并且不会重复读
 * </p>
 *
 * @author Trevor
 * @version 1.0     2017年08月10日  15点12分
 */
public class DistributedQueue<E extends Serializable> extends AbstractQueue<E>
{

    private ZKSessionFactory factory;

    public DistributedQueue(ZKSessionFactory factory)
    {
        this.factory = factory;
    }

    public Iterator<E> iterator()
    {
        throw new RuntimeException(new OperationNotSupportedException("DistributedQueue cant not provide a iterator"));
    }

    /**
     * 获取列长度
     */
    public int size()
    {
        ZKSession session;
        try
        {
            session = factory.getZKSession();
            return session.getSize();
        } catch (IOException e)
        {
            e.printStackTrace();
        }
        return -1;
    }

    /**
     * 添加元素到推列
     * @param e 实现序列化接口的元素
     * @return 是否添加成功，暂时不考虑延时重推
     */
    public boolean offer(E e)
    {
        checkNotNull(e);
        ZKSession session;
        ByteArrayOutputStream byteOut;
        ObjectOutputStream objOut = null;
        try
        {
            session = factory.getZKSession();
            byteOut = new ByteArrayOutputStream();
            objOut = new ObjectOutputStream(byteOut);
            objOut.writeObject(e);
            return session.create(byteOut.toByteArray());
        } catch (IOException e1)
        {
            e1.printStackTrace();
        } finally
        {
            try
            {
                if(objOut != null)
                    objOut.close();
            } catch (Exception ignored) {}
        }
        return false;
    }

    /**
     * 拉取对头元素
     */
    public E poll()
    {
        try
        {
            ZKSession session = factory.getZKSession();
            byte[] data = session.getFirst(true);
            if (data == null)
                return null;
            ByteArrayInputStream byteIn = new ByteArrayInputStream(data);
            ObjectInputStream objIn = new ObjectInputStream(byteIn);
            return (E) objIn.readObject();
        } catch (IOException | ClassNotFoundException e)
        {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 瞟一眼
     */
    public E peek()
    {
        try
        {
            ZKSession session = factory.getZKSession();
            byte[] data = session.getFirst(false);
            if (data == null)
                return null;
            ByteArrayInputStream byteIn = new ByteArrayInputStream(data);
            ObjectInputStream objIn = new ObjectInputStream(byteIn);
            return (E) objIn.readObject();
        } catch (IOException | ClassNotFoundException e)
        {
            e.printStackTrace();
        }
        return null;
    }

    private static void checkNotNull(Object v) {
        if (v == null)
            throw new NullPointerException();
    }

}
